package com.dean.service.impl;

import com.dean.dao.DeanToolDao;
import com.dean.entity.DeanExportInfo;
import com.dean.entity.DeanImportInfo;
import com.dean.entity.SysDbConnInfo;
import com.dean.exception.DeanRuntimeException;
import com.dean.service.DataFileImpExpService;
import com.dean.utils.DeanJdbcUtils;
import com.dean.utils.DeanSqlUtils;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVPrinter;
import org.apache.commons.csv.QuoteMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.sql.*;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

public class CsvImpExpServiceImpl implements DataFileImpExpService {
    private final Logger logger = LoggerFactory.getLogger(CsvImpExpServiceImpl.class);
    // 默认单次提交数据量为 1000;
    private static final Integer defaulCommitNum = 1000;

    private static final Integer defaultFetchSize = 1000;

    private Connection connection;

    private PreparedStatement ps;

    @Override
    /**
     * csv 文件导入功能
     * 1.目前完成单文件导入
     */
    public void imp(DeanImportInfo importInfo) {
        String filePath = importInfo.getFilePath();
        String fileName = filePath.substring(filePath.lastIndexOf("/")+1);
        String tableName = importInfo.getTableName();
        String delimiter = importInfo.getDelimiter();
        String [] fields = importInfo.getFields();
        Character quote = importInfo.getQuote();
        Boolean trim = importInfo.getTrim();
        String charset = importInfo.getCharSet();
        Integer skipHead = importInfo.getSkipHead();
        Integer skipTail = importInfo.getSkipTail();
        Integer commitNum = importInfo.getCommitNums() == null?defaulCommitNum:importInfo.getCommitNums();

        // 获取数据配置信息
        String connName = importInfo.getConnName();
        SysDbConnInfo connectInfo = DeanToolDao.getInstance().queryConnInfo(connName);
        // step 1 检验
        // step 2 构建sql
        String insertSql = DeanSqlUtils.buildInsertSql(tableName,fields,fileName);
        // step 3 流式读取数据并写入数据库
        // 组装数据 并保存到数据库
        List<List<String>> data = new ArrayList<List<String>>();

        // 读取文件行数
        long count = -1;
        try {
            count =  Files.lines(Paths.get(filePath)).count();
            logger.info("读取到文件{},记录数为{}",filePath,count);
            // 检查是否需要进行文件导入 文件记录数小于或等于跳过的记录数 则不进行处理
            if(count <= (skipHead+skipTail)){
                logger.warn("文件:{},记录数为:{},跳过前{}行,跳过后{}行,需要导入的数量为:{},注意配置及文件是否正确",fileName,count,skipHead,skipTail,count - skipHead - skipTail);
                return;
            }
        } catch (IOException e) {
            logger.error(e.getMessage(),e);
            throw DeanRuntimeException.throwException(996, e.getMessage());
        }
        // 不跳过标题行 自己根据实际情况丢弃行
        CSVFormat csvFormat = CSVFormat.RFC4180.builder().setQuoteMode(QuoteMode.MINIMAL).setTrim(trim).
                setSkipHeaderRecord(false).setQuote(quote).setDelimiter(delimiter).build();

        try(CSVParser parser = csvFormat.parse(new InputStreamReader(new FileInputStream(filePath),charset))){
            AtomicLong  atomicCount= new AtomicLong(0);
            long finalCount = count;
            parser.stream().forEach(
                    row->{
                        logger.debug("读取到文件内容:{}",row);
                        if(row.getRecordNumber()<=skipHead||row.getRecordNumber()>(finalCount - skipTail)){
                            logger.info("跳过row:{}",row);
                            return;
                        }
                        atomicCount.getAndIncrement();
                        data.add(row.toList());
                        if(atomicCount.get() % commitNum ==0){
                            saveDb(insertSql,data,connectInfo);
                            if(finalCount == atomicCount.get()){
                                connectInfo.setClose(true);
                            }
                            logger.debug("已导入[{}]行数据,完成时间[{}]",atomicCount.get(), LocalDateTime.now());
                            data.clear();
                        }
                    }
            );
            if(atomicCount.get()%commitNum >0){
                saveDb(insertSql,data,connectInfo);
                logger.debug("已导入[{}]行数据,完成时间[{}]",atomicCount.get(),LocalDateTime.now());
            }
            logger.info("文件{}已完成导入,文件总行数:{},导入行数:{},表头行数:{},表尾行数:{}",filePath,finalCount,atomicCount.get(),skipHead,skipTail);
        } catch (IOException exception) {
            logger.error("读取csv文件出错:",exception.getMessage(),exception);
            throw DeanRuntimeException.throwException(996, exception.getMessage());
        }

    }

    @Override
    public void exp(DeanExportInfo exportInfo) {
        String fileName = exportInfo.getFilePath();
        String delimiter = exportInfo.getDelimiter();
        String sql = exportInfo.getQuerySql();
        String connName = exportInfo.getConnName();
        Integer fetchSize = exportInfo.getFetchSize()==null?defaultFetchSize:exportInfo.getFetchSize();
        String charSet = exportInfo.getCharSet();
        logger.debug("文件[{}]已开始导出,分割符[{}],sql[{}],connName[{}],fetchSize[{}],开始时间[{}]",fileName,delimiter,sql,connName,fetchSize, LocalDateTime.now());
        CSVFormat csvFormat = CSVFormat.RFC4180.builder().setDelimiter(delimiter).setQuoteMode(QuoteMode.MINIMAL).setTrailingDelimiter(false).build();
        SysDbConnInfo connInfo = DeanToolDao.getInstance().queryConnInfo(connName);
        List<List<String>> datas = new ArrayList<List<String>>();
        List<String> header = new ArrayList<String>();
        try(CSVPrinter printer = csvFormat.print(Paths.get(fileName), Charset.forName(charSet));
            Connection connection = DeanJdbcUtils.getConnection(connInfo);) {
            PreparedStatement ps = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY,ResultSet.CONCUR_READ_ONLY);
            ps.setFetchSize(fetchSize);
            ps.setFetchDirection(ResultSet.FETCH_REVERSE);
            ResultSet rs = ps.executeQuery();
            ResultSetMetaData resMetaData = rs.getMetaData();
            // 表头处理
            int columnCount = resMetaData.getColumnCount();
            for(int i = 0;i < columnCount; i++){
                header.add(resMetaData.getColumnName(i+1));
            }
            datas.add(header);
            int counter = 1;
            // 处理数据
            while (rs.next()){
                List<String> data = new ArrayList<String>();
                for (int i = 0; i < columnCount; i++) {
                    data.add(rs.getString(i+1));
                }
                datas.add(data);
                counter++;
                if(counter%fetchSize == 0){
                    printer.printRecords(datas);
                    printer.flush();
                    logger.debug("共写入[{}]条数据,完成时间[{}]",counter, LocalDateTime.now());
                    datas.clear();
                }
            }
            if(datas.size()>0){
                printer.printRecords(datas);
                printer.flush();
                logger.debug("共写入[{}]条数据,完成时间[{}]",counter, LocalDateTime.now());
            }
            logger.debug("文件[{}]已完成导出,完成时间[{}] bye bye ......",fileName, LocalDateTime.now());

        } catch (IOException e) {
            logger.error("生成csv文件失败,失败原因:{}",e.getMessage(),e);
            throw DeanRuntimeException.throwException(996,e.getMessage());
        } catch (SQLException throwables) {
            logger.error("生成csv文件失败,失败原因:{}",throwables.getMessage(),throwables);
            throw DeanRuntimeException.throwException(996,throwables.getMessage());
        } catch (ClassNotFoundException e) {
            logger.error("生成csv文件失败,失败原因:{}",e.getMessage(),e);
            throw DeanRuntimeException.throwException(996,e.getMessage());
        }
    }

    /**
     * 保存结果到数据库
     * @param sql
     * @param datas
     * @param connectInfo
     */
    private  void saveDb(String sql, List<List<String>> datas, SysDbConnInfo connectInfo) {
        try{
            if (connection == null) {
                connection = DeanJdbcUtils.getConnection(connectInfo);
                connection.setAutoCommit(false);
                ps = DeanJdbcUtils.getPreparedStatement(connection, sql);
            }
            for (List<String> data : datas) {
                int size = data.size();
                for (int i = 0; i < size; i++) {
                    ps.setString(i + 1, data.get(i));
                }
                ps.addBatch();
            }
            ps.executeBatch();
            ps.clearBatch();
            connection.commit();
            // DeanJdbcUtils.batchInsert(connInfo,sql,data);
        } catch (ClassNotFoundException e) {
            logger.error(e.getMessage(),e);
            DeanRuntimeException.throwException(996,e.getMessage());
        } catch (SQLException throwables) {
            logger.error(throwables.getMessage(),throwables);
            try {
                connectInfo.setClose(true);
                connection.rollback();
            } catch (SQLException e) {
                logger.error(e.getMessage(),e);
            }
            DeanRuntimeException.throwException(996,throwables.getMessage());
        }finally {
            // 关闭连接
            if(connectInfo.getClose()){
                DeanJdbcUtils.close(ps,connection);
                connection = null;
            }
        }

    }
}
